package org.jeecg.modules.mqtt;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.jeecg.common.exception.JeecgBootException;
import org.jeecg.modules.alarmRecording.service.IAlarmRecordingService;
import org.jeecg.modules.alarmRecording.service.impl.AlarmRecordingServiceImpl;
import org.jeecg.modules.chargeElectricity.entity.ChargeElectricityDetail;
import org.jeecg.modules.chargeElectricity.service.IChargeElectricityService;
import org.jeecg.modules.chargeElectricity.service.impl.ChargeElectricityServiceImpl;
import org.jeecg.modules.consumptionElectricity.entity.ConsumptionElectricity;
import org.jeecg.modules.consumptionElectricity.mapper.ConsumptionElectricityMapper;
import org.jeecg.modules.consumptionElectricity.service.IConsumptionElectricityService;
import org.jeecg.modules.consumptionElectricity.service.impl.ConsumptionElectricityServiceImpl;
import org.jeecg.modules.emissionCarbon.entity.EmissionCarbon;
import org.jeecg.modules.emissionCarbon.service.IEmissionCarbonService;
import org.jeecg.modules.emissionCarbon.service.impl.EmissionCarbonServiceImpl;
import org.jeecg.modules.emissionCoal.entity.EmissionCoal;
import org.jeecg.modules.emissionCoal.service.IEmissionCoalService;
import org.jeecg.modules.emissionCoal.service.impl.EmissionCoalServiceImpl;
import org.jeecg.modules.energyMeterMapping.entity.EnergyMeterMapping;
import org.jeecg.modules.energyMeterMapping.service.IEnergyMeterMappingService;
import org.jeecg.modules.energyMeterMapping.service.impl.EnergyMeterMappingServiceImpl;
import org.jeecg.modules.errorLog.service.IErrorService;
import org.jeecg.modules.errorLog.service.impl.ErrorServiceImpl;
import org.jeecg.modules.temperature.entity.Temperature;
import org.jeecg.modules.temperature.service.ITemperatureService;
import org.jeecg.modules.temperature.service.impl.TemperatureServiceImpl;
import org.jeecg.modules.thresholdPlan.entity.ThresholdPlanDetail;
import org.jeecg.modules.thresholdPlan.service.IThresholdPlanService;
import org.jeecg.modules.thresholdPlan.service.impl.ThresholdPlanServiceImpl;
import org.jetbrains.annotations.NotNull;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.function.Consumer;

import static cn.hutool.core.date.DateUtil.*;

@Slf4j
public class MyMQTTCallback implements MqttCallbackExtended {
    //手动注入
    private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);

    private MyMQTTClient myMQTTClient;

    public MyMQTTCallback(MyMQTTClient myMQTTClient) {
        this.myMQTTClient = myMQTTClient;
    }

    private static final ConsumptionElectricityMapper consumptionElectricityMapper = SpringUtils.getBean(ConsumptionElectricityMapper.class);

    private final IConsumptionElectricityService consumptionElectricityService = SpringUtils.getBean(ConsumptionElectricityServiceImpl.class);

    private final IEmissionCarbonService emissionCarbonService = SpringUtils.getBean(EmissionCarbonServiceImpl.class);
    private final IChargeElectricityService chargeElectricityService = SpringUtils.getBean(ChargeElectricityServiceImpl.class);

    private final IEmissionCoalService emissionCoalService = SpringUtils.getBean(EmissionCoalServiceImpl.class);

    private static final IEnergyMeterMappingService energyMeterMappingService = SpringUtils.getBean(EnergyMeterMappingServiceImpl.class);

    private static final IErrorService errorService = SpringUtils.getBean(ErrorServiceImpl.class);

    private final ITemperatureService temperatureService = SpringUtils.getBean(TemperatureServiceImpl.class);

    private final IThresholdPlanService thresholdPlanService = SpringUtils.getBean(ThresholdPlanServiceImpl.class);

    private final IAlarmRecordingService alarmRecordingService = SpringUtils.getBean(AlarmRecordingServiceImpl.class);


    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
        long reconnectTimes = 1;
        while (true) {
            try {
                if (MyMQTTClient.getClient().isConnected()) {
                    //判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
                    log.warn("mqtt reconnect success end  重新连接  重新订阅成功");
                    return;
                }
                reconnectTimes+=1;
                log.warn("mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
                MyMQTTClient.getClient().reconnect();
            } catch (MqttException e) {
                log.error("mqtt断连异常", e);
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
            }
        }
    }

    /**
     * @param topic
     * @param mqttMessage
     * @throws Exception
     * subscribe后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
//        log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
        //发布消息主题
        if (topic.contains("nygl/acrel/adw300w-4G/mqtt/analog")){
            try {
                // 使用FastJSON解析JSON字符串为JSONObject
                JSONObject jsonObject = JSON.parseObject(new String(mqttMessage.getPayload(), StandardCharsets.UTF_8));
                /// 获取data数组
                JSONArray dataArray = jsonObject.getJSONArray("data");
                // 获取第一个data元素中的point数组
                JSONArray pointArray = dataArray.getJSONObject(0).getJSONArray("point");
                //你自己的业务接口
                insertCmdResults(pointArray);
            } catch (Exception e){
                log.error("messageArrived方法出现异常, 异常信息: ", e);
                errorService.saveError(e.getMessage(), String.valueOf(e.getCause()));
            }
        }
        if(topic.contains("nygl/usr/G771/mqtt/")){
            try {
                String gatewayId = topic.substring(topic.length() - 15);
                parseTemperature(mqttMessage, gatewayId);
            } catch (Exception e){
                log.error("messageArrived方法出现异常, 异常信息: ", e);
                errorService.saveError(e.getMessage(), String.valueOf(e.getCause()));
            }
        }
        //接收报警主题
        if (topic.equals("embed/warn")){
            Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), StandardCharsets.UTF_8));
            //你自己的业务接口
            insertPushAlarm(maps);
        }
    }

    @NotNull
    private void parseTemperature(MqttMessage mqttMessage, String gatewayId) {
        LambdaQueryWrapper<EnergyMeterMapping> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(EnergyMeterMapping::getGatewayId, gatewayId);
        EnergyMeterMapping energyMeter = energyMeterMappingService.getOne(wrapper);
        if (energyMeter == null){
            throw new JeecgBootException("设备IMEI码未找到，请检查设备编号映射");
        }
        Temperature temperature = new Temperature();
        temperature.setDept(energyMeter.getDept()).setProduct(energyMeter.getProduct())//部门;产品
                .setProductionLine(energyMeter.getProductionLine()).setPoint(energyMeter.getPoint())//生产线;点位
                .setTime(new Date()).setDevice(energyMeter.getDevice()).setGatewayId(gatewayId);//日期;设备;网关编号
        int high = 0;
        int low = 0;
        for (ThresholdPlanDetail detail : thresholdPlanService.getEnableThreshold()) {
            if (Objects.equals(energyMeter.getPoint(), detail.getPoint())){
                high = Integer.parseInt(detail.getHigh());
                low = Integer.parseInt(detail.getLow());
                break;
            }
        }
        byte[] payload = mqttMessage.getPayload();
        StringBuilder sb = new StringBuilder();
        for (byte b : payload) {
            sb.append(String.format("%02X", b));
        }
        String result = sb.substring(6, sb.length() - 4);
        if (result.contains("FE0C")) {
            errorService.saveError("当前温度为-50.0，请检查设备在线状态，设备IMEI：" + gatewayId);
        }
        int step = 4; // 每个新字符串的长度
        int len = result.length();
        List<String> list = new ArrayList<>(); // 用来存放新字符串的列表
        for (int i = 0; i < len; i += step) {
            int end = Math.min(i + step, len); // 计算每个新字符串的终止位置
            String subStr = result.substring(i, end); // 截取子字符串
            double decimal = (double) ((short) Integer.parseInt(subStr, 16)) / 10;
            if (decimal > high){
                alarmRecordingService.addRecord("温度高于预警值", energyMeter.getPoint(), energyMeter.getDept());
            }
            if (decimal < low){
                alarmRecordingService.addRecord("温度低于预警值", energyMeter.getPoint(), energyMeter.getDept());
            }
            list.add(String.valueOf(decimal)); // 截取子字符串并添加到列表
        }
        temperature.setTemperatureA(list.get(0)).setTemperatureB(list.get(1)).setTemperatureC(list.get(2));
        temperatureService.save(temperature);
    }

    private void insertCmdResults(JSONArray pointArray) throws Exception {
        JSONObject point = pointArray.getJSONObject(1);
        int d = point.getInteger("id");
        if (1 == d){
            handleFirstMessage(pointArray);
        }
//        if (d == 36){
//            return;
//        }
    }

    private void insertPushAlarm(Map maps) {

    }

    private void handleFirstMessage(JSONArray pointArray) {
        String year = String.valueOf(year(new Date()));//当前年份
        String month = String.valueOf(month(new Date()) + 1);//当前月份
        String day = String.valueOf(dayOfMonth(new Date()));//当前日
        int hour = hour(new Date(), true);//当前时

        EmissionCarbon coefficientCarbon = emissionCarbonService.getCoefficient();
        BigDecimal dianBian = coefficientCarbon.getDianbian();//获取当前碳排放电转化变量
        EmissionCoal coefficientCoal = emissionCoalService.getCoefficient();
        BigDecimal dianBian2 = coefficientCoal.getDianbian();//获取当前标煤排放电转化变量
        List<ChargeElectricityDetail> programme = chargeElectricityService.getProgramme();
        BigDecimal price = getPrice(programme, hour);//获取当前电价
        BigDecimal lastPower = null;

        ConsumptionElectricity consumptionElectricity = new ConsumptionElectricity();
        String hourStr = String.valueOf(hour);
        Map<Integer, Consumer<String>> propertySetters = new HashMap<>();
        propertySetters.put(1, consumptionElectricity::setVoltageA);//A相电压
        propertySetters.put(2, consumptionElectricity::setVoltageB);//B相电压
        propertySetters.put(3, consumptionElectricity::setVoltageC);//C相电压
        propertySetters.put(4, consumptionElectricity::setVoltageAb);//AB线电压
        propertySetters.put(5, consumptionElectricity::setVoltageBc);//BC线电压
        propertySetters.put(6, consumptionElectricity::setVoltageCa);//CA线电压
        propertySetters.put(7, consumptionElectricity::setCurrentA);//A相电流
        propertySetters.put(8, consumptionElectricity::setCurrentB);//B相电流
        propertySetters.put(9, consumptionElectricity::setCurrentC);//C相电流
        propertySetters.put(10, consumptionElectricity::setPowerA);//A相功率
        propertySetters.put(11, consumptionElectricity::setPowerB);//B相功率
        propertySetters.put(12, consumptionElectricity::setPowerC);//C相功率
        propertySetters.put(13, consumptionElectricity::setPowerTotal);//总功率
        propertySetters.put(22, consumptionElectricity::setFactorA);//A相功率因数
        propertySetters.put(23, consumptionElectricity::setFactorB);//B相功率因数
        propertySetters.put(24, consumptionElectricity::setFactorC);//C相功率因数
        propertySetters.put(25, consumptionElectricity::setFactorTotal);//总功率因数

        // 遍历point数组中的每个元素
        for (int i = 0; i < pointArray.size(); i++) {//遍历JSON数据
            JSONObject pointObject = pointArray.getJSONObject(i);//每一项JSON数据
            int id = pointObject.getInteger("id");//获取当前项数据id
            String val = String.valueOf(pointObject.get("val"));//获取当前项数据值
            Consumer<String> propertySetter = propertySetters.get(id);//根据当前项数据id获取set方法
            if (propertySetter != null) {
                propertySetter.accept(val);
            } else if (id == 0) {//当前项数据id为0,当前项数据值为电表编号
                LambdaQueryWrapper<EnergyMeterMapping> wrapper = Wrappers.lambdaQuery();
                wrapper.eq(EnergyMeterMapping::getEnergyMeterId, val);
                EnergyMeterMapping energyMeter = energyMeterMappingService.getOne(wrapper);//查询当前电表编号数据
                if (energyMeter == null){
                    throw new JeecgBootException("电表id未找到，请检查电表编号映射");
                }
                consumptionElectricity.setDept(energyMeter.getDept()).setProduct(energyMeter.getProduct())//部门;产品
                        .setProductionLine(energyMeter.getProductionLine()).setPoint(energyMeter.getPoint())//生产线;点位
                        .setDate(new Date()).setYear(year).setMonth(month).setDay(day).setHour(hourStr)//日期;年月日时
                        .setDevice(energyMeter.getDevice()).setEnergyMeterId(val);//设备;电表编号
                lastPower = getLastPowerFromDatabase(val);//根据当前电表编号获取上一次电表读数
            } else if (id == 28) {//当前项数据为28,当前项数据值为电表读数
                BigDecimal currentPower = new BigDecimal(val);//当前电表读数
                BigDecimal consumption = BigDecimal.ZERO;//耗电量
                if (lastPower != null){//如果上一次电表读数不为空
                    consumption = currentPower.subtract(lastPower);//耗电量为当前电表读数-上次电表读数
                }
                BigDecimal cost = price.multiply(consumption).setScale(5, RoundingMode.HALF_UP);//电费
                BigDecimal carbonEmission = dianBian.multiply(consumption).setScale(5, RoundingMode.HALF_UP);//碳排放
                BigDecimal coalEmission = dianBian2.multiply(consumption).setScale(5, RoundingMode.HALF_UP);//标煤排放
                consumptionElectricity.setConsumption(consumption).setTotalPower(currentPower)
                        .setCost(cost).setCarbonEmission(carbonEmission).setCoalEmission(coalEmission);//其余数据
            }
        }
//        log.info("即将向电耗表consumption_electricity中插入数据: {}", consumptionElectricity);
        consumptionElectricityService.save(consumptionElectricity);
    }

    private BigDecimal getPrice(List<ChargeElectricityDetail> programme, int hour) {
        ChargeElectricityDetail chargeElectricityDetail = programme.stream()
                .filter(ced -> {
                    if (ced.getStartTime() <= ced.getEndTime()) {
                        return ced.getStartTime() <= hour && hour <= ced.getEndTime();
                    } else {
                        return ced.getStartTime() <= hour || hour <= ced.getEndTime();
                    }
                })
                .findFirst()
                .orElse(null);
        if (chargeElectricityDetail != null) {
            return chargeElectricityDetail.getPrice();
        } else {
            return BigDecimal.ZERO;
        }
    }

    private static BigDecimal getLastPowerFromDatabase(String energyMeterId) {
        // 实现从数据库中获取最后一次的耗电量
        // 根据你的数据库类型和表结构来编写相应的查询逻辑，获取最后一次的耗电量
        QueryWrapper<ConsumptionElectricity> wrapper = new QueryWrapper<>();
        wrapper.eq("energy_meter_id", energyMeterId).orderByDesc("date").last("LIMIT 1");
        ConsumptionElectricity consumptionElectricity = consumptionElectricityMapper.selectOne(wrapper);
        if(consumptionElectricity == null || consumptionElectricity.getTotalPower() == null){
            errorService.saveError("未读取到上次电表读数（如第一次安装电表请忽略）");
            return BigDecimal.ZERO;
        }
        return consumptionElectricity.getTotalPower();
    }

    /**
     *连接成功后的回调 可以在这个方法执行 订阅主题  生成Bean的 MqttConfiguration方法中订阅主题 出现bug
     *重新连接后  主题也需要再次订阅  将重新订阅主题放在连接成功后的回调 比较合理
     * @param reconnect
     * @param serverURI
     */
    @Override
    public  void  connectComplete(boolean reconnect,String serverURI){
        log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");
        //订阅主题
        myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic2, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic3, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic4, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic5, 1);
    }

    /**
     * 消息到达后
     * subscribe后，执行的回调函数
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */
    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//        log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }

}
